with bussiness as (select recordid,
                          billcode,
                          scantype,
                          inputsite,
                          scantime,
                          uploadtime,
                          scansitecode,
                          tracepushed,
                          b.network_type,
                          if(scantype = '快件揽收', b.agent_code, null) agent_code,
                          if(scantype = '快件揽收', b.agent_name, null) agent_name,
                          if(scantype = '快件揽收', scansitecode, null) taking_site_code,
                          if(scantype = '快件揽收', inputsite, null)    taking_site,
                          if(scantype = '快件揽收', scantime, null)     taking_scantime
                   from jms_dwd.dwd_tab_barscan_business_base_hi a
                            left join
                        jms_dim.dim_network_whole_massage b
                        on a.scansitecode = b.code
                   where date(substr(dt, 1, 10)) between date_add('{{ execution_date | cst_ds }}', -6)
                       and '{{ execution_date | cst_ds }}'
                     and dt < '{{ execution_date | hour_add(-1) | cst_hour }}'
                     and nvl(tracepushed, '') <> '1'),
     taking as (select *
                from (select *,
                             row_number() over (
                                 partition by waybill_no
                                 order by
                                     scan_time
                                 ) rn
                      from jms_dwd.dwd_tab_barscan_taking_base_hi
                      where date(substr(dt, 1, 10)) between date_add(date(substr('{{ execution_date | cst_ds }}', 1, 10)), -6 )
                          and date(substr('{{ execution_date | cst_ds }}', 1, 10))
                        and dt < '{{ execution_date | hour_add(-1) | cst_hour }}') detail
                where rn = 1),
     center_arrival as (select *
                        from (select waybill_no,
                                     scan_site,
                                     scan_site_code,
                                     scan_time,
                                     scan_type,
                                     row_number() over (
                                         partition by waybill_no,scan_site
                                         order by
                                             scan_time
                                         )      rn,
                                     '发件扫描' join_scan_type
                              from jms_dwd.dwd_tab_barscan_centerarrival_base_hi
                              where date(substr(dt, 1, 10)) between date_add(date(substr('{{ execution_date | cst_ds }}', 1, 10)), -6 )
                                  and date(substr('{{ execution_date | cst_ds }}', 1, 10))
                                and dt < '{{ execution_date | hour_add(-1) | cst_hour }}') detail
                        where rn = 1),
     center_send as (select *
                     from (select waybill_no,
                                  scan_site,
                                  scan_site_code,
                                  scan_time,
                                  scan_type,
                                  row_number() over (
                                      partition by waybill_no,scan_site
                                      order by
                                          scan_time
                                      )     rn,
                                  scan_type join_scan_type
                           from jms_dwd.dwd_tab_barscan_centersend_base_hi
                           where date(substr(dt, 1, 10)) between date_add(date(substr('{{ execution_date | cst_ds }}', 1, 10)), -6 )
                               and date(substr('{{ execution_date | cst_ds }}', 1, 10))
                             and dt < '{{ execution_date | hour_add(-1) | cst_hour }}') detail
                     where rn = 1),
     center_scan as (select *
                     from center_arrival
                     union all
                     select *
                     from center_send)


insert
overwrite
table
jms_dm.dm_tab_business_monitor_detail_dt
partition
(
dt
)
select *, date(false_taking_scantime) date_time, date(false_taking_scantime) dt
from (select waybill_no,
             network_type,
             scan_site_code,
             scan_type,
             scan_site,
             scan_time,
             upload_time,
             max(agent_code) over (partition by waybill_no)                             agent_code,
             max(agent_name) over (partition by waybill_no)                             agent_name,
             max(taking_site_code) over (partition by waybill_no)                       false_taking_site_code,
             max(taking_site) over (partition by waybill_no)                            false_taking_site,
             max(taking_scantime) over (partition by waybill_no)                        false_taking_scantime,
             if(scan_type = '快件揽收', scan_time, null)                                false_taking_time,
             if(scan_type = '发件扫描' and network_type = 4, scan_time, null)                                false_send_time,
             real_taking_time,
             real_taking_type,
             real_taking_site,
             real_taking_site_code,
             real_send_type,
             real_send_site_code,
             real_send_site,
             real_send_time,
             send_diff_time,
             taking_diff_time,
             order.customer_code,
             order.customer_name,
             order.order_source_code,
             order.order_source_name,
             order.subscribe_source_code,
             order.subscribe_source_name,
             if(nvl(send_diff_time, 0) > 24 and nvl(send_diff_time, 0) <= 48 and scan_type = '发件扫描' and network_type = 4, 1, 0)     is_send_over_one_day,
             if(nvl(send_diff_time, 0) > 48 and scan_type = '发件扫描' and network_type = 4, 1, 0)                                      is_send_over_two_day,
             if(nvl(taking_diff_time, 0) > 12 and nvl(taking_diff_time, 0) <= 24, 1, 0) is_taking_over_half_day,
             if(nvl(taking_diff_time, 0) > 24 and nvl(taking_diff_time, 0) <= 48, 1, 0) is_taking_over_one_day,
             if(nvl(taking_diff_time, 0) > 48, 1, 0)                                    is_taking_send_over_two_day
      from (select waybill_no,
                   scan_type,
                   scan_site,
                   scan_time,
                   upload_time,
                   scan_site_code,
                   network_type,
                   agent_code,
                   agent_name,
                   taking_site_code,
                   taking_site,
                   taking_scantime,
                   real_taking_time,
                   real_taking_type,
                   real_taking_site,
                   real_taking_site_code,
                   if(real_send_site_code = scan_site_code or network_type <> 4, null, real_send_type) real_send_type,
                   if(real_send_site_code = scan_site_code or network_type <> 4, null,
                      real_send_site_code)                                                             real_send_site_code,
                   if(real_send_site_code = scan_site_code or network_type <> 4, null, real_send_site) real_send_site,
                   if(real_send_site_code = scan_site_code or network_type <> 4, null, real_send_time) real_send_time,
                   floor(
                               (unix_timestamp(
                                        if(real_send_site_code = scan_site_code or network_type <> 4 or
                                           scan_type not in ('发件扫描', '中心到件')
                                            , null
                                            , nvl(real_send_time,
                                                  from_unixtime(unix_timestamp(current_timestamp()) - 3600 * 2))
                                            )
                                    ) - unix_timestamp(scan_time)
                                   ) / 3600
                       )                                                                               send_diff_time,
                   floor(
                               if(scan_type = '快件揽收',
                                  unix_timestamp(
                                          nvl(real_taking_time,
                                              from_unixtime(unix_timestamp(current_timestamp()) - 3600 * 2))
                                      ) - unix_timestamp(scan_time), null) / 3600
                       )                                                                               taking_diff_time
            from (select a.billcode                                                                        waybill_no,
                         a.scantype                                                                        scan_type,
                         a.inputsite                                                                       scan_site,
                         a.scantime                                                                        scan_time,
                         a.uploadtime                                                                      upload_time,
                         a.scansitecode                                                                    scan_site_code,
                         a.network_type                                                                    network_type,
                         a.agent_code,
                         a.agent_name,
                         a.taking_site_code,
                         a.taking_site,
                         a.taking_scantime,
                         b.scan_time                                                                       real_taking_time,
                         b.scan_type                                                                       real_taking_type,
                         b.input_site                                                                      real_taking_site,
                         b.scan_site_code                                                                  real_taking_site_code,
                         send.scan_type                                                                    real_send_type,
                         send.scan_site_code                                                               real_send_site_code,
                         send.scan_site                                                                    real_send_site,
                         send.scan_time                                                                    real_send_time,
                         if(unix_timestamp(send.scan_time) - unix_timestamp(a.scantime) < 0 or
                            a.scansitecode = send.scan_site_code, 99999999999, unix_timestamp(send.scan_time) -
                                                                               unix_timestamp(a.scantime)) send_time_diff,
                         row_number() over (partition by a.billcode,a.scantype,a.inputsite order by if(
                                             unix_timestamp(send.scan_time) - unix_timestamp(a.scantime) < 0 or
                                             a.scansitecode = send.scan_site_code, 99999999999,
                                             unix_timestamp(send.scan_time) - unix_timestamp(a.scantime))) rn
                  from bussiness a
                           left join taking b
                                     on a.billcode = b.waybill_no
                                         and a.scantype = b.scan_type
                           left join center_scan send
                                     on a.billcode = send.waybill_no
                                         and a.scantype = send.join_scan_type) detail
            where rn = 1) detail
               left join jms_dwd.dwd_yl_oms_oms_order_hf
      order
      on detail.waybill_no = order.waybill_id
      and date(substr(order.dt, 1, 10)) 
      between date_add(date(substr('{{ execution_date | cst_ds }}', 1, 10)), -10 + 1)
      and date(substr('{{ execution_date | cst_ds }}', 1, 10))
      and order.dt < '{{ execution_date | hour_add(-1) | cst_hour }}' 
     ) detail where date(false_taking_scantime) between date_add(date('{{ execution_date | cst_ds }}'), -3) and '{{ execution_date | cst_ds }}'
    distribute by dt
   , PMOD(HASH(RAND()), 8);










  
